package com.hp.bon.sgw.core;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.hp.bon.sgw.Constants;
import com.hp.bon.sgw.Constants.ResultCode;
import com.hp.bon.sgw.bean.CapabilityCtrlBean;
import com.hp.bon.sgw.bean.SGWStatQueueBean;
import com.hp.bon.sgw.bean.SysParamBean;
import com.hp.bon.sgw.service.ConfigListener;
import com.hp.bon.sgw.service.ConfigService;
import com.hp.bon.sgw.start.SpringContext;
import com.hp.bon.sgw.util.Comutils;

/**
 * 正在处理中，还未完成处理的请求消息队列 包括节点发出的请求，以及收到的请求。
 * 
 * @author wuzhih
 * 
 */
public class InprogressMessageQueue implements ConfigListener {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = LoggerFactory.getLogger(InprogressMessageQueue.class);
	/**
	 * 当前允许的消息处理最大并发数
	 */
	private volatile int maxInprogressCount = 10000;
	/**
	 * DCCProgress 超时时间（秒）
	 */
	private volatile int dccProgressTimeoutSeconds = 30;

	/**
	 * 守护线程扫描的间隔时间
	 */
	private volatile int watchDogScanSecondes = 1;
	/**
	 * 队列容量告警阀值
	 */
	private volatile int queueWarnThreshold = 66;

	private ConfigService confServ;

	/**
	 * 保存当前正在处理中的消息状态
	 */
	public static ConcurrentHashMap<String, CallParam> msgProgressMap = new ConcurrentHashMap<String, CallParam>();

	WatchDogThread watchDogThread;

	/**
	 * 所关心的运行期变化的配置数据的Key
	 */
	private final Set<String> monitorConfKeys = new HashSet<String>();
	private volatile long prevQuequeWarnedTime;

	public InprogressMessageQueue() {
		monitorConfKeys.add(SysParamBean.KEY_AGREEMENT_Timeout_LinkWait);
		monitorConfKeys.add("InprogressMessageQueue.Max");
		monitorConfKeys.add("InprogressMessageQueue.WarnThreshold");
		monitorConfKeys.add("InprogressMessageQueue.WatchDogScanSconds");
	}

	static InprogressMessageQueue instance;

	public static InprogressMessageQueue getInstance() {
		if (instance == null) {
			instance = SpringContext.getApplicationContext().getBean(InprogressMessageQueue.class);
		}
		return instance;
	}

	/**
	 * 初始化过程
	 */
	public void init() {
		loadConfParams();
		watchDogThread = new WatchDogThread();
		watchDogThread.start();
	}

	@Override
	public void onConfigUpdated(String updatedKey) {
		loadConfParams();
	}

	private void loadConfParams() {
		String paramName = SysParamBean.KEY_AGREEMENT_Timeout_LinkWait;
		dccProgressTimeoutSeconds = confServ.getIntValue(paramName, 60);
		logger.info("conf param " + paramName + " value " + dccProgressTimeoutSeconds);
		paramName = "InprogressMessageQueue.Max";
		maxInprogressCount = confServ.getIntValue(paramName, 10000);
		logger.info("conf param " + paramName + " value " + maxInprogressCount);
		paramName = "InprogressMessageQueue.WarnThreshold";
		queueWarnThreshold = confServ.getIntValue(paramName, 80);
		logger.info("conf param " + paramName + " value " + queueWarnThreshold);
		paramName = "InprogressMessageQueue.WatchDogScanSeconds";
		watchDogScanSecondes = confServ.getIntValue(paramName, 1);
		logger.info("conf param " + paramName + " value " + watchDogScanSecondes);

	}

	/**
	 * 队列使用率 0-100之间
	 * 
	 * @return
	 */
	public double getQueueUsage() {
		return 100 * (msgProgressMap.size() + 0.0) / maxInprogressCount;
	}

	/**
	 * 取队列最大长度
	 * 
	 * @return
	 */
	public int getMaxInprogressCount() {
		return maxInprogressCount;
	}

	/**
	 * 第一个元素是队列当前长度，第二个元素是所有请求的数据字节长度总和
	 * 
	 * @return
	 */
	public int getQueueSize() {
		return msgProgressMap.size();
	}

	public void addRequest(CallParam callParam) {
		if (callParam.getTimeOutTime() == 0L) {
			// 如果没有设置超时时间
			int timeout = dccProgressTimeoutSeconds;
			CapabilityCtrlBean bean = callParam.getCapabilityBean();
			if ((bean != null) && (bean.getTimeoutDropMsg() != 0)) {
				timeout = bean.getTimeoutDropMsg();
			}
			callParam.setTimeOutTime(System.currentTimeMillis() + timeout * 1000);
		}

		msgProgressMap.put(callParam.getSessionId(), callParam);
	}

	/**
	 * 将原有的某个msg process移除
	 * 
	 * @param oldMsgID
	 */
	public CallParam removeProgress(String oldMsgID) {
		return msgProgressMap.remove(oldMsgID);
	}

	@Autowired(required = true)
	public void setConfServ(ConfigService confServ) {
		this.confServ = confServ;
		confServ.addConfigListener(this);
	}

	public SGWStatQueueBean getQueueStatusBean() {
		SGWStatQueueBean bean = new SGWStatQueueBean();
		bean.setMaxQueue(this.maxInprogressCount);
		bean.setUsedQueue(getQueueSize());
		bean.setThreshold(this.queueWarnThreshold);
		long warnedTime = this.prevQuequeWarnedTime;
		if (warnedTime != 0) {
			bean.setPrevWarnedTime(Comutils.formatTime(warnedTime, Comutils.GEN_DATA_FORMAT));
		} else {
			bean.setPrevWarnedTime("");
		}
		return bean;
	}
	/**
	 * 得到当前队列中各种请求的Pending的消息数目
	 * 
	 * @return 队列长度
	 */
	public int[] getPendingCounts() {
		int[] totalCounts = new int[3];
		// for (CallParam progrss : this.msgProgressMap.values()) {
		// int indx = 0;
		// switch (progrss.getSgwGateWayType()) {
		// case Constants.SGW_GATEWAY_TYPE_DCC:
		// indx = 0;
		// break;
		// case Constants.SGW_GATEWAY_TYPE_WS:
		// indx = 1;
		// break;
		//
		// case Constants.SGW_GATEWAY_TYPE_SCSP:
		// indx = 2;
		// break;
		// default:
		// indx = -1;
		// }
		// if (indx != -1) {
		// totalCounts[indx] += 1;
		// }
		// }
		return totalCounts;
	}

	/**
	 * 得到当前队列中的能力数量统计信息
	 * 
	 * @return
	 */
	public int getCurCapsSum(Map<String, Integer> sumMap) {
		int totalCount = 0;
		for (CallParam progrss : msgProgressMap.values()) {
			String cap = progrss.getCapability();
			Integer value = sumMap.get(cap);
			value = (value == null) ? 1 : value++;
			sumMap.put(cap, value);
			totalCount++;
		}
		return totalCount;

	}

	public void shutdown() {
		watchDogThread.interrupt();
	}

	class WatchDogThread extends Thread {
		public WatchDogThread() {
			setName("InprogressMessageQueue-WatchDogThread");
			this.setDaemon(true);
		}

		public void run() {
			while (!Thread.interrupted()) {
				checkQueue();
				try {
					Thread.sleep(watchDogScanSecondes * 1000);
				} catch (Exception e) {

				}
			}
			logger.info("shutdowned");
		}

		private void checkQueue() {
			try {
				int queueLen = 0;
				Iterator<Map.Entry<String, CallParam>> itor = msgProgressMap.entrySet().iterator();
				long curTime = System.currentTimeMillis();
				while (itor.hasNext()) {
					queueLen++;
					Map.Entry<String, CallParam> entry = itor.next();
					CallParam msgPrgrss = entry.getValue();
					if (msgPrgrss.getTimeOutTime() < curTime) {// 已经超时
						// 从队列中删除,并调用Process处理应答
						itor.remove();
						logger.warn(Constants.ResultCode.QueueMsg_Timeout + "|found message process timeout,sessionId=" + msgPrgrss.getMessageUniqueID() + ",serviceCtxID=" + msgPrgrss.getServiceContextId());
						msgPrgrss.setRetCode(ResultCode.BON_TIMEOUT);
						//描述是哪个网元超时
						msgPrgrss.setResultMsg("BON timeout:"+msgPrgrss.getToNode().getHostId());
						msgPrgrss.getProcessor().endProcess(msgPrgrss);
					}
				}
				// 检查是否队列容量超阀值
				if (queueLen * queueWarnThreshold / 100 > maxInprogressCount && prevQuequeWarnedTime + 5 * 60 * 1000 < curTime) {
					logger.warn(Constants.ResultCode.Queue_Over_Threshold + "|current queue reached threshold " + queueWarnThreshold + ",cur length:" + queueLen + ",max capacity:" + maxInprogressCount);
					prevQuequeWarnedTime = curTime;
				}
			} catch (Exception e) {
				e.printStackTrace();
				logger.warn("checkQueue:" + e.toString());
			}
		} 
	}

	@Override
	public Set<String> monitorConfKeys() {
		return monitorConfKeys;
	}

	public String getQueneContext() {
		StringBuilder sb = new StringBuilder();
		Iterator<Map.Entry<String, CallParam>> itor = msgProgressMap.entrySet().iterator();
		Map.Entry<String, CallParam> entry = null;

		while (itor.hasNext()) {
			entry = itor.next();
			sb.append(entry.getKey() + " = " + Comutils.formatTime(entry.getValue().getTimeOutTime(), "yyyy-MM-dd hh:mm:ss.SSS"));
			sb.append("\r\n");
		}

		return sb.toString();
	}
}